package com.sfx.messagequeue.mqServer.datacenter;

import com.sfx.messagequeue.common.consumer.Consumer;
import com.sfx.messagequeue.common.consumer.ConsumerEnv;
import com.sfx.messagequeue.common.exception.MQException;
import com.sfx.messagequeue.mqServer.VirtualHost;
import com.sfx.messagequeue.mqServer.core.MSGQueue;
import com.sfx.messagequeue.mqServer.core.Message;

import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingQueue;

/**
 * Created with IntelliJ IDEA.
 * Description: 这个类主要实现消息消息的过程
 * User: sfx
 * Date: 2023-08-09
 * Time: 15:39
 */
public class ConsumerManager {
    // VirtualHost用于 管理数据
    private final VirtualHost virtualHost;
    // 线程池用户处理/执行 "消费者" 的回调函数
    private final ExecutorService WORKER_POOL = Executors.newFixedThreadPool(4);
    // 异步扫描线程
    private Thread SCAN_THREAD = null;
    // 阻塞队列用于存放消息的 "令牌" (产生消息的队列名字)
    private final BlockingQueue<String> TOKEN_QUEUE = new LinkedBlockingQueue<>();

    public ConsumerManager(VirtualHost virtualHost) {
        // 创建出一个扫描线程,去 "令牌" 队列中 取出 令牌->队列名,然后去队列中消费消息
        SCAN_THREAD = new Thread(() -> {
            while (true) {
                try {
                    //从阻塞队列中取出令牌
                    String queueName = TOKEN_QUEUE.take();
                    MSGQueue queue = virtualHost.getMemoryDataCenter().getQueue(queueName);
                    if (queue == null) {
                        System.out.println("[ConsumerManager] 扫描线程取出队列,队列不存在 queueName=" + queueName);
                    }
                    //消费消息
                    synchronized (queue) {
                        consumeMessage(queue);
                    }
                } catch (InterruptedException e) {
                    throw new RuntimeException(e);
                }
            }
        }, "SCAN_THREAD");
        SCAN_THREAD.setDaemon(true); //设置为守护线程
        SCAN_THREAD.start();
        this.virtualHost = virtualHost;
    }

    //通知去消费消息
    public void notifyConsume(String queueName) throws InterruptedException {
        TOKEN_QUEUE.put(queueName);
    }


    /**
     * 主要实现订阅消息
     * 往队列中添加一个订阅者,收到消息后,把消息推送给对应的订阅者->消费者
     *
     * @param consumerTag 消费者的标识
     * @param queueName   要订阅消息的队列
     * @param autoAck     应答方式 为true代表自动应答(消息拿走,啥也不干)   false手动应答(需要调用basicAck方法)
     * @param consumer    回调函数, 这里主要处理当触发basicConsume方法的时候,需要触发回调函数consumer
     */
    public void addConsumer(String consumerTag, String queueName, boolean autoAck, Consumer consumer) throws MQException {
        //1. 查询对应的队列
        MSGQueue queue = virtualHost.getMemoryDataCenter().getQueue(queueName);
        if (queue == null) {
            throw new MQException("[ConsumerManager] 队列不存在! queueName=" + queueName);
        }
        //2. 创建一个订阅者-
        ConsumerEnv consumerEnv = new ConsumerEnv(consumerTag, queueName, autoAck, consumer);
        //3. 往队列中添加订阅者到队列中
        synchronized (queue) {
            queue.addConsumerEnv(consumerEnv);
            //4. 看队列中是否已经有消息了,如果有就先把队列中的消息消费掉
            int messageSize = virtualHost.getMemoryDataCenter().messageSizeByQueue(queueName);
            for (int i = 0; i < messageSize; ++i) {
                //消费消息
                consumeMessage(queue);
            }
        }
    }

    /**
     * 消费消息
     *
     * @param queue 队列
     */
    private void consumeMessage(MSGQueue queue) {
        //因为一个队列中有多个消费者
        //所以我们需要 使用 "轮询" 的方式 挑选出一个 "消费者" 来进行消费

        //1.挑选出一个消费者
        ConsumerEnv luckyDog = queue.chooseConsumerEnv();
        if (luckyDog == null) {
            //这个队列没有订阅者
            return;
        }
        //2. 从队列中取出消息
        Message message = virtualHost.getMemoryDataCenter().pollMessage(queue.getName());
        if (message == null) {
            //队列中没有消息,等待有消息了,在去执行
            System.out.println("[ConsumerManager] 队列中没有消息 queueName=" + queue.getName());
            return;
        }
        //3. 交给线程池去执行消费者的回调
        WORKER_POOL.submit(() -> {
            try {
                //3.1 执行回调函数之前,需要先保存到待确认的集合中,防止消息丢失
                virtualHost.getMemoryDataCenter().addMessageWaitAck(queue.getName(), message);
                //3.2 线程池去执行回调
                luckyDog.getConsumer().handleDelivery(luckyDog.getConsumerTag(), message.getBasicProperties(), message.getBody());
                //3.3 根据消费者的应答模式来确定什么时候把消息删除掉
                if (luckyDog.isAutoAck()) {
                    // isAutoAck 为true 代表自动应答,如果回调不抛出异常,直接将消息删除
                    // 1): 看消息是否持久化,如果持久化,从磁盘中删除
                    if (message.getDeliverMode() == 2) {
                        virtualHost.getDiskDataCenter().deleteMessage(queue, message);
                    }
                    // 2): 消息从内存中删除
                    virtualHost.getMemoryDataCenter().deleteMessage(message.getMessageId());
                    // 3): 消息从WaitAckMap中删除
                    virtualHost.getMemoryDataCenter().removeMessageWaitAck(queue.getName(), message);
                }
            } catch (Exception e) {
                System.out.println("[ConsumerManager] 消费者消费消息失败queueName=" + queue.getName());
                e.printStackTrace();
            }
        });
    }
}